package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.text.SimpleDateFormat;

/*
访客UV计算
数据源： dwd_page_log
 */
public class UniqueVisitApp {
    public static void main(String[] args) throws Exception {
        //TODO 0. 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);

        //TODO 1. 从kafka读取数据
        String  groupId = "unique_visit_app";
        String sourceTopic = "dwd_page_log";
        String sinkTopic = "dwm_unique_visit";

        //读取kafka数据
        FlinkKafkaConsumer<String> source = MyKafkaUtil.getKafkaSource(sourceTopic,groupId);
        DataStreamSource<String> jsonStream = env.addSource(source);

        //对读取的数据进行结构的转换
        DataStream<JSONObject> jsonObjStream = jsonStream.map(JSON::parseObject);
//        jsonObjStream.print("uv::::");


        // TODO 2. 核心的过滤代码
        //按照设备id(mid)进行分组, 第二个字段是key
        KeyedStream<JSONObject,String> keyedByWithMidDS = jsonObjStream.keyBy(
                jsonObject -> jsonObject.getJSONObject("common").getString("mid")
        );

        SingleOutputStreamOperator<JSONObject> filterJsonObjectDS = keyedByWithMidDS.filter(
                new RichFilterFunction<JSONObject>() {

                    //定义状态用于存放最后的访问日期
                    ValueState<String> lastVisitDataState =null;
                    SimpleDateFormat simpleDateFormat = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        simpleDateFormat= new SimpleDateFormat("yyyy-MM-dd");
                        if(lastVisitDataState == null){
                            ValueStateDescriptor<String> lastViewDateStateDescriptor =
                            new ValueStateDescriptor<String>("lastViewDateState",String.class);

                            // 统计日活，设置状态失效时间
                            StateTtlConfig  stateTtlConfig = StateTtlConfig.newBuilder(Time.days(1)).build();

                            lastViewDateStateDescriptor.enableTimeToLive(stateTtlConfig);
                            lastVisitDataState = getRuntimeContext().getState(lastViewDateStateDescriptor);
                        }
                    }

                    // 首先检查当前页面是否有标记，如果有则说明访问一定不是当前访问,上一页 肯定不是首次的进入
                    @Override
                    public boolean filter(JSONObject jsonObject) throws Exception {
                        String lastPageId= jsonObject.getJSONObject("page").getString("last_page_id");
                        if(lastPageId != null && lastPageId.length() >0){
                            return false;
                        }

                        Long ts = jsonObject.getLong("ts");
                        String logDate = simpleDateFormat.format(ts);
                        String lastViewDate = lastVisitDataState.value();


                        if(lastViewDate != null && lastViewDate.length() >0 && logDate.equals(lastViewDate)){
                            System.out.println("已访问: lastVisit: "+ lastViewDate + "|| logDate: "+ logDate);
                            return false;
                        }else{
                            System.out.println("未访问: lastVisit: "+ lastViewDate +"|| logDate: "+ logDate);
                            lastVisitDataState.update(logDate);
                            return true;
                        }
                    }
                }
        )
        // TODO ?  UID 函数的功能,视频中也没有这一行代码  后面测试, 源码注释： 指定的ID用于跨作业提交分配相同的操作符ID(例如，当从保存点启动作业时)。
                .uid("uvFilter");

        SingleOutputStreamOperator<String> dataJsonStringDS = filterJsonObjectDS.map(jsonObject -> jsonObject.toJSONString());

        dataJsonStringDS.print("uv");


        //TODO 3. 将过滤处理后的UV写入到kafka的dwm_unique_visit
        dataJsonStringDS.addSink(MyKafkaUtil.getKafkaSink(sinkTopic));









        env.execute();
    }
}
